package streamforce.crypto.cloud.subscriber;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import streamforce.Stat;
import streamforce.Tools;
import streamforce.crypto.KPAbe;
import streamforce.crypto.cloud.EPCloud;
import streamforce.crypto.event.Ciphertext;
import streamforce.crypto.event.StockEvent;
import streamforce.crypto.event.StockEventJoin;
import streamforce.crypto.policies.Policy;

/**
 * First step in the Join process, i.e. listening to
 * 	select * from StockEvent(stockId=0/1)
 * and then transform the deterministic encryption of the price field. 
 * Finally, we forward it to another stream of StockEventJoin
 */
public class CloudJoinPriceStage1 extends Policy{
	final Log log = LogFactory.getLog(CloudJoinPriceStage1.class);
	int policyId; 
	Stat stat;
	public CloudJoinPriceStage1(int pi, KPAbe cryptoHandler){
		super(cryptoHandler); 
		this.policyId = pi;
		stat = new Stat("nTransformed_join");		
	}
	
	public  void update(StockEvent se){
		byte[] ciphertext = se.getPrice().get("price_det").getCipher();
		long start = System.currentTimeMillis(); 
		byte[] trans = this.cryptoHandler.transform_join(se.getStockId(), this.policyId, ciphertext);		
		stat.update();
		Ciphertext ct = new Ciphertext(trans); 
		se.getPrice().put("price_det", ct);
		long end = System.currentTimeMillis(); 
		//log.info(String.format("got ( %d, %d )\n",se.getStockId(), se.getTimestamp())); 
		StockEventJoin sej = new StockEventJoin(se.getTimestamp(),se.getRealTime(),
				se.getStockId(), se.getPrice(), se.getVolume());		
		EPCloud.getEPServiceProvider().getEPRuntime().sendEvent(sej);				
	}		
		
}
